
package com.alibaba.datax.plugin.writer.httpwriter;

import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.math.BigInteger;
import java.net.HttpURLConnection;
import java.net.URL;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.text.SimpleDateFormat;
import java.util.*;

public class HttpWriter extends Writer {
    public static final String UTF_8 = "UTF-8";

    public static class Job extends Writer.Job {
        private static final Logger LOG = LoggerFactory
                .getLogger(Job.class);

        private Configuration originalConfig;

        @Override
        public void init() {
            this.originalConfig = super.getPluginJobConf();
            validateParameter();
        }

        private void validateParameter() {
            String url = this.originalConfig.getString(Key.URL);
            try {
                if (StringUtils.isBlank(url)||(!(url.startsWith("http://")||url.startsWith("https://")))) {
                    throw DataXException
                            .asDataXException(
                                    HttpWriterErrorCode.ILLEGAL_VALUE,
                                    String.format(
                                            "您配置的url: [%s] 为空或不是http接口地址.",
                                            url));
                }
            } catch (SecurityException se) {
                throw DataXException.asDataXException(
                        HttpWriterErrorCode.SECURITY_NOT_ENOUGH,
                        String.format("url错误 : [%s] ", url), se);
            }
        }

        @Override
        public void prepare() {
        }

        @Override
        public List<Configuration> split(int mandatoryNumber) {
            List<Configuration> writerSplitConfigs = new ArrayList<Configuration>();
            for (int i = 0; i < mandatoryNumber; i++) {
                writerSplitConfigs.add(this.originalConfig);
            }

            return writerSplitConfigs;
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }
    }

    public static class Task extends Writer.Task {
        private static final Logger LOG = LoggerFactory
                .getLogger(Task.class);

        private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");

        private Configuration writerSliceConfig;

        private boolean print;

        private String url;
        private String clientId;
        private List<String> column;
        private String clientSecret;
        private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");


        @Override
        public void init() {
            this.writerSliceConfig = getPluginJobConf();
            this.print = this.writerSliceConfig.getBool(Key.PRINT, true);
            this.url = this.writerSliceConfig.getString(Key.URL, null);
            this.clientId = this.writerSliceConfig.getString(Key.CLIENTID, null);
            this.clientSecret = this.writerSliceConfig.getString(Key.CLIENTSECRET, null);
            this.column = this.writerSliceConfig.getList(Key.COLUMN,String.class);
        }

        @Override
        public void prepare() {
        }

        @Override
        public void startWrite(RecordReceiver recordReceiver) {
            LOG.info("begin do write...");
            try {
                List<Map<String,Object>> datas = new ArrayList<>();
                Record record;
                while ((record = recordReceiver.getFromReader()) != null) {
                    datas.add(recordToMap(record));
                    if(datas.size()%100==0){
                        String rtn = httpPost(JSON.toJSONString(datas));
                        LOG.info(rtn);
                        datas.clear();
                    }
                }
                if(datas.size()>0){
                    String rtn = httpPost(JSON.toJSONString(datas));
                    LOG.info(rtn);
                }
            } catch (Exception e) {
                throw DataXException.asDataXException(HttpWriterErrorCode.RUNTIME_EXCEPTION, e);
            }
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

        private Map<String,Object> recordToMap(Record record) {
            int recordLength = record.getColumnNumber();
            if (0 == recordLength) {
                return null;
            }
            Column column;
            Map<String,Object> row = new HashMap<>();
            for (int i = 0; i < recordLength; i++) {
                column = record.getColumn(i);
                row.put(this.column.get(i),column.asString());
            }
            return row;
        }

        private String httpPost(String data) throws IOException {
            LOG.info(this.url);
            LOG.info(data);
            URL url = new URL(this.url);
            HttpURLConnection conn = (HttpURLConnection)url.openConnection();
            conn.setDoInput(true);// 允许输入
            conn.setDoOutput(true);// 允许输出
            conn.setUseCaches(false); // 不允许使用缓存
            conn.setRequestMethod("POST");
            conn.setRequestProperty("connection", "keep-alive");
            conn.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
            conn.setRequestProperty(Key.CLIENTID,this.clientId);
            String timestamp = sdf.format(new Date());
            conn.setRequestProperty(Key.TIMESTAMP,timestamp);
            String Authorization = stringToMD5(clientId+clientSecret+timestamp);
            conn.setRequestProperty(Key.AUTHORIZATION,Authorization);
            conn.getOutputStream().write(data.getBytes(UTF_8));
            conn.getOutputStream().flush();
            BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream()));
            StringBuilder sb = new StringBuilder();
            br.lines().forEach(v1->sb.append(v1).append(NEWLINE_FLAG));
            return sb.toString();
        }

        public static String stringToMD5(String plainText) {
            byte[] secretBytes = null;
            try {
                secretBytes = MessageDigest.getInstance("md5").digest(
                        plainText.getBytes(UTF_8));
            } catch (NoSuchAlgorithmException e) {
                throw new RuntimeException("没有这个md5算法！");
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            String md5code = new BigInteger(1, secretBytes).toString(16);
            for (int i = 0; i < 32 - md5code.length(); i++) {
                md5code = "0" + md5code;
            }
            return md5code;
        }
    }

}
